/**
 * Copyright (c) 2013-2022 Nikita Koksharov
 * <p>
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * <p>
 * http://www.apache.org/licenses/LICENSE-2.0
 * <p>
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.redisson;

import java.util.Arrays;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;

import org.redisson.api.RFuture;
import org.redisson.api.RLock;
import org.redisson.client.codec.LongCodec;
import org.redisson.client.codec.StringCodec;
import org.redisson.client.protocol.RedisCommands;
import org.redisson.client.protocol.RedisStrictCommand;
import org.redisson.command.CommandAsyncExecutor;
import org.redisson.pubsub.LockPubSub;

/**
 * 读写锁 - 读锁
 */
public class RedissonReadLock extends RedissonLock implements RLock {

    protected RedissonReadLock(CommandAsyncExecutor commandExecutor, String name) {
        super(commandExecutor, name);
    }


    /**
     * 获取频道名称
     */
    @Override
    String getChannelName() {
        return prefixName("redisson_rwlock", getRawName());
    }

    /**
     * 获取写锁的名字
     */
    String getWriteLockName(long threadId) {
        return super.getLockName(threadId) + ":write";
    }

    /**
     * 获取读写超时名称前缀
     */
    String getReadWriteTimeoutNamePrefix(long threadId) {
        return suffixName(getRawName(), getLockName(threadId)) + ":rwlock_timeout";
    }


    /**
     * 读锁加锁方法
     */
    @Override
    <T> RFuture<T> tryLockInnerAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {

        /**
         * KEYS = Arrays.asList(getRawName(), getReadWriteTimeoutNamePrefix(threadId))
         * KEYS[1] = getRawName()  获取锁对象时设置的参数，比如"read_write_lock"
         * KEYS[2] = getReadWriteTimeoutNamePrefix(threadId)  {read_write_lock}:UUID:threadId:rwlock_timeout
         *
         * ARGV = unit.toMillis(leaseTime), getLockName(threadId), getWriteLockName(threadId)
         * ARGV[1] = unit.toMillis(leaseTime) // 锁超时时间，默认30s
         * ARGV[2] = getLockName(threadId) // 读锁名称，UUID:ThreadId
         * ARGV[3] = getWriteLockName(threadId) // 写锁名称，UUID:threadId:write
         *
         * 首次加锁，先执行lua脚本的第一个if分支，观察一下redis的数据，其中一个有一个key为 read_write_lock结构的hash锁，
         * 包含锁的模式，加锁的线程重入次数；
         *
         * 另一个key为{redis_write_lock}:7b8aefd7-e765-4788-abb4-e4c40264eda4:76:rwlock_timeout:1 结构为String的数据
         * 主要记录的是当前线程的超市时间
         */

        return commandExecutor.syncedEval(getRawName(), LongCodec.INSTANCE, command,
                // local mode = hget redisson_read_write_lock mode
                "local mode = redis.call('hget', KEYS[1], 'mode'); " +

                // 无锁(无读锁,无写锁),  加读锁成功后 return null
                        // if (mode == false) then
                        //     hset redisson_read_write_lock mode  read
                        //     hset redisson_read_write_lock UUID:ThreadId 1
                        //     set {redisson_read_write_lock}:UUID:ThreadId:rwlock_timeout:1 1
                        //     pexpire {redisson_read_write_lock}:UUID:ThreadId:rwlock_timeout:1 30000
                        //     pexpire redisson_read_write_lock 30000
                        //     return null;
                        // end
                        "if (mode == false) then " +
                        "redis.call('hset', KEYS[1], 'mode', 'read'); " +
                        "redis.call('hset', KEYS[1], ARGV[2], 1); " +
                        "redis.call('set', KEYS[2] .. ':1', 1); " +
                        "redis.call('pexpire', KEYS[2] .. ':1', ARGV[1]); " +
                        "redis.call('pexpire', KEYS[1], ARGV[1]); " +
                        "return nil; " +
                        "end; " +


                // 锁降级, 读读兼容、读写互斥、写写互斥、写读互斥
                // 1、读锁
                // 2、写锁同时写锁的名称为当前线程
                // 加读锁成功后 return null
                        // if (mode == read ||　(mode == write && hexist redisson_read_write_lock UUID:threadId:write == 1)) then
                        //   local ind = hincrby redisson_read_wirte_lock UUID:threadId 1
                        //   set {redisson_read_write_lock}:UUID:ThreadId:rwlock_timeout:ind 1
                        //   pexpire {read_write_lock }:UUID:threadId::rwlock_timeout:ind 30000
                        //
                        //   local remainTime =  pttl "redisson_read_wirte_lock"
                        //   pexpire "redisson_read_wirte_lock" math.max(remainTime, 30)
                        //   retrun null
                        //   end

                        "if (mode == 'read') or (mode == 'write' and redis.call('hexists', KEYS[1], ARGV[3]) == 1) then " +
                        "local ind = redis.call('hincrby', KEYS[1], ARGV[2], 1); " +
                        "local key = KEYS[2] .. ':' .. ind;" +
                        "redis.call('set', key, 1); " +
                        "redis.call('pexpire', key, ARGV[1]); " +
                        "local remainTime = redis.call('pttl', KEYS[1]); " +
                        "redis.call('pexpire', KEYS[1], math.max(remainTime, ARGV[1])); " +
                        "return nil; " +
                        "end;" +

                // 最后判断条件都不成立
                // 非当前线程的写锁
                // return pttl "redisson_read_write_lock"
                        "return redis.call('pttl', KEYS[1]);",
                Arrays.<Object>asList(getRawName(), getReadWriteTimeoutNamePrefix(threadId)),
                unit.toMillis(leaseTime), getLockName(threadId), getWriteLockName(threadId));
    }


    /**
     * 读锁解锁方法
     */
    @Override
    protected RFuture<Boolean> unlockInnerAsync(long threadId) {
        String timeoutPrefix = getReadWriteTimeoutNamePrefix(threadId);
        String keyPrefix = getKeyPrefix(threadId, timeoutPrefix);

        /**
         * KEYS = Arrays.asList(getRawName(), getChannelName(), timeoutPrefix, keyPrefix)
         * KEYS[1] = getRawName()  即获取锁对象时设置的 “read_write_lock”
         * KEYS[2] = getChannelName()  订阅消息的通道，redisson_rwlock:{read_write_lock}
         * KEYS[3] = timeoutPrefix     {read_write_lock }:UUID:threadId:rwlock_timeout
         * KEYS[4] = keyPrefix         {read_write_lock }
         *
         * ARGV = LockPubSub.UNLOCK_MESSAGE, getLockName(threadId)
         * ARGV[1] = LockPubSub.UNLOCK_MESSAGE   Redis发布事件的message 0L
         * ARGV[2] = getLockName(threadId)       读锁名称，UUID:ThreadId
         * ARGV[3] = getSubscribeService().getPublishCommand()  PUBLISH
         */

        return commandExecutor.syncedEval(getRawName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,

                // local mode = hget redisson_read_write_lock mode
                "local mode = redis.call('hget', KEYS[1], 'mode'); " +

                // 无锁(无读锁,无写锁),发送订阅消息,结束
                // 无锁,return 1
                        // if (mode == false) then
                        //    PUBLISH redisson_rwlock:{redisson_read_write_lock} 0L
                        //    return 1
                        //    end
                        "if (mode == false) then " +
                        "redis.call(ARGV[3], KEYS[2], ARGV[1]); " +
                        "return 1; " +
                        "end; " +

                // 当前线程不持有锁
                // 不存在读锁情况下,return null
                        // if (hexists redisson_read_write_lock UUID:ThreadId == 0) then
                        //    return null
                        // end
                        "local lockExists = redis.call('hexists', KEYS[1], ARGV[2]); " +
                        "if (lockExists == 0) then " +
                        "return nil;" +
                        "end; " +

                // 走到这里,说明该线程持有读锁,此时此刻开始解锁操作
                // 1、 锁无重入,彻底解锁   retrun 1
                // 2、 锁有重入,局部解锁   return 0
                        // local counter = hincrby redisson_read_write_lock UUID:ThreadId -1
                        // if (counter == 0) then
                        //      hdel  redisson_read_write_lock UUID:ThreadId
                        // end
                        // del {redisson_read_write_lock}:UUID:threadId:rwlock_timeout:(counter+1)

                        // if (hlen redisson_read_write_lock > 1) then
                        //     local maxRemainTime = -3;
                        //     local keys = hkeys redisson_read_write_lock
                        //     for n,key in ipairs(keys) do
                        //         counter = tonumber(hget redisson_read_write_lock key)
                        //         if (type(counter) == 'number') then
                        //              for i=counter, 1, -1 do
                        //                  local remainTime = pttl {redisson_read_write_lock}:key:rwlock_timeout: ind
                        //                  maxRemainTime = math.max(remainTime, maxRemainTime);
                        //                  end
                        //              end
                        //      end

                        //     if maxRemainTime > 0 then
                        //        pexpire redisson_read_write_lock maxRemainTime
                        //        return 0
                        //     end

                        //     if mode == write then
                        //        return 0
                        //     end
                        // end
                        "local counter = redis.call('hincrby', KEYS[1], ARGV[2], -1); " +
                        "if (counter == 0) then " +
                        "redis.call('hdel', KEYS[1], ARGV[2]); " +
                        "end;" +
                        "redis.call('del', KEYS[3] .. ':' .. (counter+1)); " +

                        "if (redis.call('hlen', KEYS[1]) > 1) then " +
                        "local maxRemainTime = -3; " +
                        "local keys = redis.call('hkeys', KEYS[1]); " +
                        "for n, key in ipairs(keys) do " +
                        "counter = tonumber(redis.call('hget', KEYS[1], key)); " +
                        "if type(counter) == 'number' then " +
                        "for i=counter, 1, -1 do " +
                        "local remainTime = redis.call('pttl', KEYS[4] .. ':' .. key .. ':rwlock_timeout:' .. i); " +
                        "maxRemainTime = math.max(remainTime, maxRemainTime);" +
                        "end; " +
                        "end; " +
                        "end; " +
                        "if maxRemainTime > 0 then " +
                        "redis.call('pexpire', KEYS[1], maxRemainTime); " +
                        "return 0; " +
                        "end;" +
                        // 如果锁模式为 write 返回
                        "if mode == 'write' then " +
                        "return 0;" +
                        "end; " +
                        "end; " +

                // 如果走到这步，说明当前线程解锁后，这个读写锁没有任何线程持有了，这时把读写锁删除
                // 并且往对应的 channel 中发送解锁的消息
                        // del redisson_read_write_lock
                        // PUBLISH redisson_rwlock:{redisson_read_write_lock} 0L
                        // return 1;
                        "redis.call('del', KEYS[1]); " +
                        "redis.call(ARGV[3], KEYS[2], ARGV[1]); " +
                        "return 1; ",
                Arrays.<Object>asList(getRawName(), getChannelName(), timeoutPrefix, keyPrefix),
                LockPubSub.UNLOCK_MESSAGE, getLockName(threadId), getSubscribeService().getPublishCommand());
    }

    protected String getKeyPrefix(long threadId, String timeoutPrefix) {
        return timeoutPrefix.split(":" + getLockName(threadId))[0];
    }


    /**
     * 锁续期方法
     */
    @Override
    protected CompletionStage<Boolean> renewExpirationAsync(long threadId) {
        String timeoutPrefix = getReadWriteTimeoutNamePrefix(threadId);
        String keyPrefix = getKeyPrefix(threadId, timeoutPrefix);

        /**
         * KEYS = Arrays.asList(getRawName(), keyPrefix)
         * KEYS[1] = getRawName() // 即获取锁对象时设置的 “read_write_lock”
         * KEYS[2] = keyPrefix() // 即 {read_write_lock}
         *
         * ARGV = internalLockLeaseTime, getLockName(threadId)
         * ARGV[1] = internalLockLeaseTime // 默认 30 s
         * ARGV[2] = getLockName(threadId) // 读锁名称，UUID:ThreadId
         */
        return evalWriteAsync(getRawName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
                // 获取 read_write_lock UUID:ThreadId 的值，也就是获取读锁的重入次数
                // hget read_write_lock UU
                "local counter = redis.call('hget', KEYS[1], ARGV[2]); " +
                        "if (counter ~= false) then " +
                        // 对 read_write_lock 重新设置过期时间 默认30s
                        "redis.call('pexpire', KEYS[1], ARGV[1]); " +
                        // 如果 read_write_lock 里面field个数大于2个 获取所有的key
                        "if (redis.call('hlen', KEYS[1]) > 1) then " +
                        "local keys = redis.call('hkeys', KEYS[1]); " +
                        // 对 read_write_lock 里的 key 进行遍历
                        "for n, key in ipairs(keys) do " +
                        // 依旧是取值为重入次数的 key
                        "counter = tonumber(redis.call('hget', KEYS[1], key)); " +
                        "if type(counter) == 'number' then " +
                        // 倒序遍历
                        "for i=counter, 1, -1 do " +
                        // 重新设置过期时间
                        // pexpire {read_write_lock }:UUID:ThreadId:rwlock_timeout:i 30000
                        "redis.call('pexpire', KEYS[2] .. ':' .. key .. ':rwlock_timeout:' .. i, ARGV[1]); " +
                        "end; " +
                        "end; " +
                        "end; " +
                        "end; " +

                        "return 1; " +
                        "end; " +
                        "return 0;",
                Arrays.<Object>asList(getRawName(), keyPrefix),
                internalLockLeaseTime, getLockName(threadId));
    }

    @Override
    public Condition newCondition() {
        throw new UnsupportedOperationException();
    }


    /**
     * KEYS = Arrays.asList(getRawName(), getChannelName())
     * KEYS[1] = getRawName() 即获取锁对象时设置的 “read_write_lock”
     * KEYS[2] = getChannelName() 订阅消息的通道，redisson_rwlock:{read_write_lock}
     *
     * ARGV = LockPubSub.UNLOCK_MESSAGE, getSubscribeService().getPublishCommand()
     * ARGV[1] = LockPubSub.UNLOCK_MESSAGE   Redis发布事件的message 0L
     * ARGV[2] = getSubscribeService().getPublishCommand()  PUBLISH
     */

    /**
     * 强制解锁
     */
    @Override
    public RFuture<Boolean> forceUnlockAsync() {
        cancelExpirationRenewal(null);
        return commandExecutor.syncedEvalWithRetry(getRawName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
                // hget read_write_lock mode == read
                // 只要读锁
                "if (redis.call('hget', KEYS[1], 'mode') == 'read') then " +
                        // del read_write_lock
                        "redis.call('del', KEYS[1]); " +
                        // PUBLISH redisson_rwlock:{read_write_lock}  0L
                        "redis.call(ARGV[2], KEYS[2], ARGV[1]); " +
                        "return 1; " +
                        "end; " +
                        "return 0; ",
                Arrays.asList(getRawName(), getChannelName()),
                LockPubSub.UNLOCK_MESSAGE, getSubscribeService().getPublishCommand());
    }

    @Override
    public boolean isLocked() {
        RFuture<String> future = commandExecutor.writeAsync(getRawName(), StringCodec.INSTANCE, RedisCommands.HGET, getRawName(), "mode");
        String res = get(future);
        return "read".equals(res);
    }

}
